package com.sui.bigdata.rtcadmin.async;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.sui.bigdata.rtcadmin.constant.JobConstant;
import com.sui.bigdata.rtcadmin.repository.mapper.JobConfigMapper;
import com.sui.bigdata.rtcadmin.repository.model.JobConfig;
import com.sui.bigdata.rtcadmin.util.AlarmUtils;
import com.sui.bigdata.rtcadmin.util.RedisUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;

import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_PREFIX;
import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_SUFFIX;
import static com.sui.bigdata.rtcadmin.util.SendMsgUtils.sendJobRestartFailMsg;


/**
 * @author YongChen
 * @date 2020/1/6 10:45
 * @description
 * @email yong_chen@sui.com
 */
@Component
public class JobAlarm implements Runnable {

    protected static final Logger logger = LoggerFactory.getLogger(JobAlarm.class);

    private static String ID = "job";

    private final Long ttl = 24 * 60 * 60L;

    private JobConfig jobConfig;

    private String jobStatus;

    @Value("${oneday.max.restart.time}")
    private Integer onedayMaxRestartTime;

    @Value("${alarm.msg.url}")
    private String alarmMsgUrl;

    @Autowired
    private RedisUtils redisUtils;

    @Autowired
    private AlarmUtils alarmUtils;

    @Autowired
    private JobConfigMapper jobConfigMapper;

    public void setJobConfig(JobConfig jobConfig) {
        this.jobConfig = jobConfig;
    }

    public void setJobStatus(String jobStatus) {
        this.jobStatus = jobStatus;
    }

    @Override
    public void run() {
        String alarmMsgUrlNew = alarmMsgUrl;
        if (StringUtils.isNotBlank(jobConfig.getAlarmGroupPhone())) {
            alarmMsgUrlNew = alarmMsgUrl + jobConfig.getAlarmGroupPhone();
        }
        String lockKey = jobConfig.getJobName() + "_rs";
        String lockValue = UUID.randomUUID().toString();
        try {
            if (redisUtils.setScheduler(lockKey, lockValue, 120)) {
                SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
                String todayDate = formatter.format(new Date(System.currentTimeMillis()));

                String jobName = jobConfig.getJobName();
                String key = ID+jobName + todayDate;
                Integer todayRestartTimes = redisUtils.getInteger(key);

                if (todayRestartTimes < onedayMaxRestartTime) {
                    alarmUtils.restartJob(jobName, alarmMsgUrlNew, jobConfig,
                            CustomConfiguration.getString(FLINK_WEB_PREFIX) + jobConfig.getAppId() + FLINK_WEB_SUFFIX, jobStatus, todayRestartTimes + 1);
                    redisUtils.set(key, todayRestartTimes + 1, ttl);
                    logger.warn("{} was restarted for the {} time today", jobName, todayRestartTimes + 1);
                } else {
                    logger.warn("{} has restarted 3 times.", jobName);
                    // 修改任务状态为HANG状态
                    jobConfig.setStatus(JobConstant.HANG);
                    sendJobRestartFailMsg(alarmMsgUrlNew, jobName, jobConfig.getDeveloper(), jobStatus,
                            CustomConfiguration.getString(FLINK_WEB_PREFIX) + jobConfig.getAppId() + FLINK_WEB_SUFFIX,
                            todayRestartTimes);
                    jobConfigMapper.updateStatusByJobName(jobName, JobConstant.HANG, JobConstant.JOB_IS_ONLINE);
                }
            }
        } catch (Exception e) {
            logger.error("restartTask {} exception.", jobConfig.getJobName(), e);
        }finally {
            // 任务重启逻辑结束之后就要把锁给释放。
            redisUtils.remove(lockKey, lockValue);
        }

    }

    public static JobAlarm build(JobAlarm jobAlarm, JobConfig jobConfig, String jobStatus) {
        jobAlarm.setJobConfig(jobConfig);
        jobAlarm.setJobStatus(jobStatus);
        return jobAlarm;
    }
}
